package org.genitus.forceps.hbase;

import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.github.benmanes.caffeine.cache.RemovalListener;
import org.apache.hadoop.hbase.KeepDeletedCells;
import org.apache.hadoop.hbase.TableExistsException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.RegionSplitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

public class BulkProcessor {
    private static final Logger LOG = LoggerFactory.getLogger(HBaseClient.class);

    private LoadingCache<String, BufferedMutator> mutatorLoadingCache;

    private final byte[] columnFamily;
    private final Connection connection;

    private BulkProcessor(Connection connection, byte[] columnFamily,Long buffSize, BufferedMutator.ExceptionListener exceptionListener) {
        this.columnFamily = columnFamily;
        this.connection = connection;
        this.mutatorLoadingCache = Caffeine.newBuilder()
                .expireAfterAccess(2, TimeUnit.MINUTES)
                .removalListener((RemovalListener<String, BufferedMutator>) (key, value, cause) -> {
                    try {
                        assert value != null;
                        value.flush();
                        value.close();
                        LOG.info("remove {} mutator from cache success according {}",key,cause.name());
                    } catch (Exception e) {
                        LOG.error("remove {} mutator from cache has exception: {}",key,e.getMessage());
                    }
                })
                .build(key-> {
                    LOG.info("load {} mutator to cache success",key);
                    createTableIfAbsent(TableName.valueOf(key));
                    return getMutator(TableName.valueOf(key),buffSize,exceptionListener);
                });
    }

    /**
     * get mutator
     * @param tableName         HBase table name
     * @param buffSize          mutator buff size
     * @param exceptionListener exception listener
     * @return                  bufferedMutator
     * @throws IOException      HBase connection exception
     */
    private BufferedMutator getMutator(TableName tableName,Long buffSize,BufferedMutator.ExceptionListener exceptionListener) throws IOException {
        BufferedMutatorParams params = new BufferedMutatorParams(tableName)
                .writeBufferSize(buffSize)
                .setWriteBufferPeriodicFlushTimeoutMs(2000)
                .listener(exceptionListener);
        return connection.getBufferedMutator(params);
    }

    /**
     * create table if table is absent
     *
     * @param  tableName    HBase table name
     * @throws IOException  HBase connection exception
     */
    private void createTableIfAbsent(TableName tableName) throws IOException {
        try(Admin hBaseAdmin = connection.getAdmin()) {
            if (hBaseAdmin.isTableAvailable(tableName)) return;

            ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamily)
                    .setDataBlockEncoding(DataBlockEncoding.NONE)
                    .setBloomFilterType(BloomType.ROW)
                    .setCompactionCompressionType(Compression.Algorithm.SNAPPY)
                    .setKeepDeletedCells(KeepDeletedCells.FALSE)
                    .setBlockCacheEnabled(true)
                    .setTimeToLive(2147483647)
                    .setBlocksize(65536)
                    .setInMemory(false)
                    .setMinVersions(0)
                    .setMaxVersions(1)
                    .build();

            TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
                    .setColumnFamily(familyDescriptor)
                    .setReplicationScope(0)
                    .build();

            RegionSplitter.SplitAlgorithm splitAlgorithm = RegionSplitter.newSplitAlgoInstance(
                    hBaseAdmin.getConfiguration(), "HexStringSplit"
            );

            hBaseAdmin.createTable(tableDescriptor,splitAlgorithm.split(600));
        } catch (TableExistsException e) {
            LOG.info("HBase table {} has been created by other instance",tableName.getNameAsString());
        }
    }

    /**
     * Write log with sid to HBase with specific table col
     *
     * @param rowKey        rowKey
     * @param quf           qualifier
     * @param data          data
     * @throws IOException  HBase connection exception
     */
    public void write(String table, String rowKey, String quf, byte[] data) throws IOException {
        BufferedMutator bufferedMutator = mutatorLoadingCache.get(table);
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(columnFamily, Bytes.toBytes(quf), data);
        put.setDurability(Durability.SKIP_WAL);
        assert bufferedMutator!=null;
        bufferedMutator.mutate(put);
    }

    public void close() {
        mutatorLoadingCache.invalidateAll();
    }

    /**
     * A builder used to create a build an instance of a bulk processor.
     */
    public static class Builder {
        private final Connection connection;
        private final byte[] columnFamily;
        private final BufferedMutator.ExceptionListener listener;

        private long buffSize = 1024 * 1024 * 6;

        private Builder(Connection connection, String columnFamily,BufferedMutator.ExceptionListener listener){
            this.connection = connection;
            this.columnFamily = Bytes.toBytes(columnFamily);
            this.listener = listener;
        }

        public BulkProcessor.Builder setBuffSize(long buffSize) {
            this.buffSize = buffSize;
            return this;
        }

        public BulkProcessor build() throws IOException {
            return new BulkProcessor(connection,columnFamily,buffSize,listener);
        }
    }

    public static BulkProcessor.Builder builder(Connection connection, String columnFamily, BufferedMutator.ExceptionListener listener) {
        Objects.requireNonNull(connection, "connection");
        Objects.requireNonNull(columnFamily, "columnFamily");
        Objects.requireNonNull(listener, "listener");
        return new BulkProcessor.Builder(connection,columnFamily,listener);
    }

}
